package cn.stylefeng.guns.rabbitmq.Consumer;

import cn.stylefeng.guns.modular.yingbao.entity.SysSell;
import cn.stylefeng.guns.modular.yingbao.entity.SysUserMessage;
import cn.stylefeng.guns.modular.yingbao.handler.BuildingHandler;
import cn.stylefeng.guns.modular.yingbao.handler.SysSellHandler;
import cn.stylefeng.guns.modular.yingbao.service.SysSellService;
import cn.stylefeng.guns.modular.yingbao.service.SysUserMessageService;
import cn.stylefeng.guns.modular.yingbao.service.impl.SysBuildingServiceImpl;
import cn.stylefeng.guns.websocket.WebSocketServerEndpoint;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;

import javax.websocket.Session;
import java.io.IOException;
import java.util.Date;
import java.util.Map;


/**
 * @author jj
 * @date 2019/12/9 15:28
 */

@Configuration
@EnableRabbit
@Slf4j
public class ConsumerConfig implements RabbitListenerConfigurer {

    @Autowired
    private ConnectionFactory connectionFactory;
    @Autowired
    SysSellHandler sysSellHandler;
    @Autowired
    BuildingHandler buildingHandler;

    @Autowired
    SysUserMessageService sysUserMessageService;

    @Autowired
    private WebSocketServerEndpoint webSocketServerEndpoint;

    Map<String, Session> livingSessions = WebSocketServerEndpoint.livingSessions;
    @Bean
    public DefaultMessageHandlerMethodFactory handlerMethodFactory() {
        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
        factory.setMessageConverter(new MappingJackson2MessageConverter());
        return factory;
    }

    @Bean
    public SimpleMessageListenerContainer messageContainer(@Qualifier("queueContract") Queue contract, @Qualifier("queueQualification") Queue qualification) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(contract, qualification);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        //设置确认模式手工确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                byte[] body = message.getBody();
                log.info("receive msg : " + new String(body));
                try {
                    String data = new String(body);
                    JSONObject data1 = JSONObject.parseObject(data);
                    Session session = livingSessions.get(data1.getString("id"));
                    if (session==null){
                        SysUserMessage sysUserMessage=new SysUserMessage();
                        sysUserMessage.setUserid(data1.getLong("id"));
                        sysUserMessage.setMessage(data1.getString("message"));
                        sysUserMessage.setCreateTime(new Date());
                        sysUserMessage.setStuats(0);
                        sysUserMessageService.save(sysUserMessage);
                        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                        return;
                    }
                    webSocketServerEndpoint.sendMessageToAll(new String(body));
                    //确认消息成功消费
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (IOException e) {
                    log.error("消息推送前台出错：" + e.getMessage() + "/r/n重新发送");
                    //重新发送
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                }
            }
        });
        return container;
    }

    /**
     *订单超时未支付
     * @param contract
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer disMessageContainer(@Qualifier("queueHander") Queue contract) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(contract);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        //设置确认模式手工确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                byte[] body = message.getBody();
                log.info("receive msg : " + new String(body));
                try {
                    String data = new String(body);
                    JSONObject data1 = JSONObject.parseObject(data);
                    Long sellId = data1.getLong("message");
                    //处理订单
                    sysSellHandler.payErrorHandler(sellId);
                    //确认消息成功消费
                    //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (IOException e) {
                    log.error("消息推送前台出错：" + e.getMessage() + "/r/n重新发送");
                    //重新发送
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                }
            }
        });
        return container;
    }

    /**
     *订单超时未确定
     * @param contract
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer buildMessageContainer(@Qualifier("queueAffirmHander") Queue contract) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(contract);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        //设置确认模式手工确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                byte[] body = message.getBody();
                log.info("receive msg : " + new String(body));
                try {
                    String data = new String(body);
                    JSONObject data1 = JSONObject.parseObject(data);
                    Long sellId = data1.getLong("message");
                    //处理订单
                    sysSellHandler.affirmHandler(sellId);
                    //确认消息成功消费
                    //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (IOException e) {
                    log.error("消息推送前台出错：" + e.getMessage() + "/r/n重新发送");
                    //重新发送
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                }
            }
        });
        return container;
    }
    /**
     *建仓完成
     * @param contract1,contract2,contract3,contract4
     * @return
     */
    @Bean
    public SimpleMessageListenerContainer affirmMessageContainer(@Qualifier("queueBuildinghanderXs") Queue contract1,
                                                                 @Qualifier("queueBuildinghanderCu") Queue contract2,
                                                                 @Qualifier("queueBuildinghanderZh") Queue contract3,
                                                                 @Qualifier("queueBuildinghanderGa") Queue contract4,
                                                                 @Qualifier("queueBuildinghanderXsVip") Queue contract5,
                                                                 @Qualifier("queueBuildinghanderCuVip") Queue contract6,
                                                                 @Qualifier("queueBuildinghanderZhVip") Queue contract7,
                                                                 @Qualifier("queueBuildinghanderGaVip") Queue contract8) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueues(contract1,contract2,contract3,contract4,contract5,contract6,contract7,contract8);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        //设置确认模式手工确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(new ChannelAwareMessageListener() {
            @Override
            public void onMessage(Message message, Channel channel) throws Exception {
                byte[] body = message.getBody();
                log.info("receive msg : " + new String(body));
                try {
                    String data = new String(body);
                    JSONObject data1 = JSONObject.parseObject(data);

                    //处理订单
                    buildingHandler.buildingHandler(data1);
                    //确认消息成功消费
                    //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (IOException e) {
                    log.error("消息推送前台出错：" + e.getMessage() + "/r/n重新发送");
                    //重新发送
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                }
            }
        });
        return container;
    }



    @Override
    public void configureRabbitListeners(RabbitListenerEndpointRegistrar rabbitListenerEndpointRegistrar) {
        rabbitListenerEndpointRegistrar.setMessageHandlerMethodFactory(handlerMethodFactory());
    }
}
